package com.cn.rpcframe.zookeeperoperation;

 

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.cn.rpcframe.constant.Constant;
import com.cn.rpcframe.registeranddiscover.ServiceRegisterAndDiscovery;
 
 

public class ZookeeperServicImpl implements ServiceRegisterAndDiscovery {

	  private static final Logger logger = LoggerFactory.getLogger(ZookeeperServicImpl.class);

	 private CountDownLatch latch = new CountDownLatch(1);
	 private ZooKeeper zk = null;
	 private String serviceAddress;
	 private List<String> dataList;
	 public ZookeeperServicImpl(String serviceAddress){
		 this.serviceAddress=serviceAddress;
		 this.zk = this.connectServer(serviceAddress);
		 if(this.zk!=null){
			 watchNode();
			 createRootNode();
		 }
	 }
	 
	 
	public boolean register(String serviceName ) {
		// TODO Auto-generated method stub
        if(this.zk!=null){
        	
        	createNode(serviceName);
        	
        	return true;
        }
        return false;
	}
/**
 * 创建节点
 * @param data
 * @return
 */
	public boolean createNode( String data) {
        if(zk == null){   
        	return false;
        }
        try {
            byte[] bytes = data.getBytes();
            String[] serviceInfo = data.split("-");
            String path = zk.create(Constant.ZK_REGISTRY_PATH+"/"+serviceInfo[0], bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            logger.debug("create zookeeper node ({} => {})", path, data);
        } catch (KeeperException e) {
            logger.error("", e);
        }
        catch (InterruptedException ex){
            logger.error("", ex);
        }
        
		return true;
	}
	/**
	 * 创建根节点
	 * @param zk
	 */
    private void createRootNode( ){
        try {
            Stat s = zk.exists(Constant.ZK_REGISTRY_PATH, false);
            if (s == null) {
                zk.create(Constant.ZK_REGISTRY_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (KeeperException e) {
            logger.error(e.toString());
        } catch (InterruptedException e) {
            logger.error(e.toString());
        }
    }
 
	public ZooKeeper connectServer(String serverHost) {
		// TODO Auto-generated method stub
		ZooKeeper zk = null;
	        try {
	            zk = new ZooKeeper(serverHost, Constant.ZK_SESSION_TIMEOUT, new Watcher() {
	                @Override
	                public void process(WatchedEvent event) {
	                    if (event.getState() == Event.KeeperState.SyncConnected) {
	                        latch.countDown();
	                    }
	                }
	            });
	            latch.await();
	        } catch (IOException | InterruptedException e) {
	            logger.error("", e);
	            return null;
	        }
	         return zk;
	}

	@Override
	public String discovery(String serviceName ) {
		// TODO Auto-generated method stub
		if(this.dataList!=null){
			if(this.dataList.size()==1){
				return this.dataList.get(0);
			}
			int size = this.dataList.size();
			String data = dataList.get(
					ThreadLocalRandom.current().nextInt(size));
			return data;
		}
		
		return null;
	}
	/**
	 * 观察zookeeper节点的变化情况
	 * @param zk
	 */
    private void watchNode( ) {
        try {
            List<String> nodeList = zk.getChildren(Constant.ZK_REGISTRY_PATH, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getType() == Event.EventType.NodeChildrenChanged) {
                        watchNode( );
                    }
                }
            });
            List<String> dataList = new ArrayList<>();
            for (String node : nodeList) {
                byte[] bytes = zk.getData(Constant.ZK_REGISTRY_PATH + "/" + node, false, null);
                dataList.add(new String(bytes));
            }
            logger.debug("node data: {}", dataList);
            this.dataList = dataList;

            logger.debug("Service discovery triggered updating connected server node.");
           
        } catch (KeeperException | InterruptedException e) {
            logger.error("", e);
        }
    }
	 
    
}
